package gnss_plugs

/***
  实时数据流写入文件库
*/

import (
	"dgo/goutils/gnss"
	"dgo/goutils/utils"
	"fmt"
	"io"
	"math"
	"sync"
	"sync/atomic"
	"time"
)

/*
**

	时候数据写入类
*/
type GnssRinexObs3xWriterAfter struct {
	gnssRinexObs3xWriterBase
	writing_flag int32 // 正在写入
	recv_epoch_n int32 // 接收到的历元数, 不包含过滤的历元
	Id           string
	MountPoint   string

	msmInput *gnss.RtcmMsmEpochInput

	// 当前历元缓存块
	curEpochCacheRec *gnssRinexObs3xWriterEpochCacheRec

	cacheEpochReclst *utils.SyncCycleList // 等待写入缓存块
	cacheEpochNum    int                  // 设定缓存历元数量, 满了后, 推入待写入列表, 比如一秒1个, 设定60个, 则一分钟会投递到待写入列表, 默认60 * 5 最小10
	writerMap        utils.GSyncMapGroup  // 文件名->真正的文件名

	// 最后写入开始的时间
	lastWriteCacheT    time.Time
	lastWriteFileSuccT time.Time // 最后写入文件成功的时间
	lastWriteInfo      string    // 最后写入统计信息
	writefile_size     int64     // 写入文件的字节数

	//  写入文件时, 如果rinex数据超过该值则 会写入一次文件
	writeCacheBlockSize uint64

	firstRecvRtcmT time.Time // 第一次接收到Rtcm数据的时间, 系统时间

	FilterNaviSys string // 只存储匹配的系统  GREC
	FixInterval   int64  // 固定采样间隔 单位：秒  5s,  mod(t, 5) = 0

	closeW sync.WaitGroup

	// 检测系统重入bug
	lastgid          uint64
	checkworkingflag int32
	lastdebuginfo    string
	DebuginfoFunc    func() string

	Filter_Msm_StartT time.Time // 过滤数据开始时间
	Filter_Msm_EndT   time.Time // 过滤数据截止时间
}

func NewGnssRinexObs3xWriterAfter() *GnssRinexObs3xWriterAfter {
	rval := &GnssRinexObs3xWriterAfter{
		Id:                  utils.RandKeyString(4),
		writeCacheBlockSize: utils.SIZE_MB * 2,
		cacheEpochNum:       60 * 5,
	}
	rval.Filefmt = ""
	rval.writeCacheBlockSize = utils.SIZE_MB * 2
	return rval
}

func (this *GnssRinexObs3xWriterAfter) GetFirstRecvRtcmT() time.Time {
	return this.firstRecvRtcmT
}

func (this *GnssRinexObs3xWriterAfter) GetLastSuccWriteFileT() time.Time {
	return this.lastWriteFileSuccT
}

func (this *GnssRinexObs3xWriterAfter) GetWriteFileSize() int64 {
	return this.writefile_size
}

func (this *GnssRinexObs3xWriterAfter) GetLastWriteWarning() string {
	return this.lastWringWaring
}

func (this *GnssRinexObs3xWriterAfter) GetRecvEpochNum() int32 {
	return this.recv_epoch_n
}

func (this *GnssRinexObs3xWriterAfter) GetLastWriteInfo() string {
	return this.lastWriteInfo
}

/*
*

	触发一次写入
	单线程执行
*/
func (this *GnssRinexObs3xWriterAfter) writeCache2File() {
	lst := this.cacheEpochReclst
	if lst == nil {
		return
	}
	if !atomic.CompareAndSwapInt32(&this.writing_flag, 0, 1) {
		return
	}
	if lst.Count() == 0 {
		return
	}
	this.lastWriteCacheT = time.Now()
	this.closeW.Add(1)
	defer func() {
		this.closeW.Done()
		this.writing_flag = 0
	}()
	for {
		if ok, rec := lst.PeekFirst(); ok {
			v := utils.GetFirstFromVars(rec)
			if v == nil {
				this.lastWringWaring = fmt.Sprintf("[%s]cacheBlock中第一个数据非法, 为nil", utils.NowString())
				lst.RemoveFirst() // 非法的数据
				continue
			}
			cacheRec := v.(*gnssRinexObs3xWriterEpochCacheRec)
			w := this.checkWriter(cacheRec) // 获取写入器
			if w == nil || w.IO == nil {
				break
			}
			if w.write_size == 0 || w.rinexHeader == nil {
				w.rinexHeader = this.checkRinexHeader(cacheRec.ObsFileName)
			}
			if w.rinexHeader == nil {
				break

			}

			t1 := time.Now()
			s1 := this.writefile_size
			if this.tryWriteBlock(w, cacheRec) { // 处理完成
				lst.RemoveFirst()

				{
					// 处理写入统计信息
					t2 := time.Now().Sub(t1)
					n1 := this.writefile_size - s1
					if t2.Seconds() == 0 {
						this.lastWriteInfo = fmt.Sprintf("[%s] size:%s, t:%.2f (s)", utils.NowString(), utils.HumanFilesize(n1), t2.Seconds())
					} else {
						this.lastWriteInfo = fmt.Sprintf("[%s] size:%s, t:%.2f (s), speed:%s/s", utils.NowString(), utils.HumanFilesize(n1), t2.Seconds(), utils.HumanFilesize(int64(float64(n1)/t2.Seconds())))
					}
				}
			} else {
				this.lastWringWaring = fmt.Sprintf("[%s]tryWriteBlock写入失败", utils.NowString())
				break
			}
		} else {
			break
		}

	}
}

func (this *GnssRinexObs3xWriterAfter) checkPostCurrentCache() bool {
	rec := this.curEpochCacheRec
	if rec != nil {
		if this.cacheEpochReclst == nil {
			this.cacheEpochReclst = utils.NewSyncCycleList()
			this.cacheEpochReclst.ConfigMaxSize(5)
		}
		this.cacheEpochReclst.Push(this.curEpochCacheRec)
		this.curEpochCacheRec = nil
		return true
	}
	return false
}

/*
**

	将历元投入当前缓存, 并判断是否需要开一个新的缓存记录
*/
func (this *GnssRinexObs3xWriterAfter) checkInputEpoch(epoch *gnss.RtcmMsmEpoch) {
	rec := this.curEpochCacheRec
	if rec != nil {
		strFileN := utils.ParseTimeFmt(this.Filefmt, epoch.EpochTime)
		if strFileN != rec.ObsFileName {
			if this.checkPostCurrentCache() {
				this.requestWrite2File()
			}
		}

	}

	if rec == nil {
		this.curEpochCacheRec = newEpochCacheRec(this.cacheEpochNum)
		rec = this.curEpochCacheRec
		rec.ObsFileName = utils.ParseTimeFmt(this.Filefmt, epoch.EpochTime)
		rec.BeginObsTime = epoch.EpochTime
	}

	atomic.AddInt32(&this.recv_epoch_n, 1)
	rec.epochList.Push(epoch)
	if rec.epochList.Count() == int32(this.cacheEpochNum) { // 队列满了
		if this.checkPostCurrentCache() {
			this.requestWrite2File()
		}
	}
	return
}

func (this *GnssRinexObs3xWriterAfter) writeRinexHeader2Writer(w io.Writer, temp_signHeader string, temp_starttime, end_time time.Time) error {
	// first line
	str := gnss.PrintRinexFV(3.04, 2, 9) + FPrintRinexStrV("OBSERVATION DATA", 27) + FPrintRinexStrV("M (MIXED)", 13)
	str = utils.AddSuffixForWidth(str, 60, " ")
	str += "RINEX VERSION / TYPE\n"

	str += utils.AddSuffixForWidth(RINEX_TransVer, 20, " ") +
		utils.AddSuffixForWidth(RINEX_CopyRight, 20, " ") +
		utils.AddSuffixForWidth(temp_starttime.Format("20060102 150405")+" UTC", 20, " ") +
		"PGM / RUN BY / DATE\n"

	str += temp_signHeader
	if !temp_starttime.IsZero() {
		str += "  " + temp_starttime.Format("2006    01    02    15    04   05") + ".0000000     GPS         TIME OF FIRST OBS\n"
		str += "  " + end_time.Format("2006    01    02    15    04   05") + ".0000000     GPS         TIME OF LAST OBS\n"
	}
	str += utils.AddSuffixForWidth("", 60, " ") + "END OF HEADER\n"

	_, err := w.Write([]byte(str))
	return err
}

func (this *GnssRinexObs3xWriterAfter) Close() {
	this.notifyDone()
	time.Sleep(time.Millisecond * 100)
	this.closeW.Wait()
}

func (this *GnssRinexObs3xWriterAfter) requestWrite2File() {
	if this.cacheEpochReclst == nil || this.cacheEpochReclst.Count() == 0 {
		return
	}
	this.writeCache2File()
}

func (this *GnssRinexObs3xWriterAfter) notifyDone() {
	if this.msmInput != nil {
		this.msmInput.Close()
	}
	this.checkPostCurrentCache()
	this.requestWrite2File()
}

/*
*

	返回false 表示需要过滤掉的数据
*/
func (this *GnssRinexObs3xWriterAfter) checkFilterMsm(obs *gnss.RtcmMsm) bool {
	if this.FixInterval > 0 {
		f_msec := float64(obs.MsmTime.UnixNano()) / 1e6 // 毫秒 采样率有 10Hz的数据
		i_msec := int64(math.Floor(f_msec + 0.1))
		v := i_msec % (this.FixInterval * 1000)
		if v != 0 {
			return false
		} else {
			return true
		}
	}
	return true
}

func (this *GnssRinexObs3xWriterAfter) checkFilterNaviSys(obs *gnss.RtcmMsm) bool {
	return true
}

func (this *GnssRinexObs3xWriterAfter) WriteDelaySecs() float64 {
	if this.writing_flag == 0 {
		return -1
	}

	return time.Now().Sub(this.lastWriteCacheT).Seconds()
}

func (this *GnssRinexObs3xWriterAfter) Reset() {

}

func (this *GnssRinexObs3xWriterAfter) checkRinexHeader(matchfilename string) (rval *gnss.RinexHeader) {
	lst := this.cacheEpochReclst
	if lst == nil {
		return nil
	}

	j := 0
	lst.Range(func(args ...interface{}) bool {
		rec := args[0].(*gnssRinexObs3xWriterEpochCacheRec)
		if len(matchfilename) == 0 {
			matchfilename = rec.ObsFileName
		}
		if rec.ObsFileName == matchfilename {
			rec.epochList.Range(func(idx int, value interface{}, removeit *bool) bool {
				epoch := value.(*gnss.RtcmMsmEpoch)
				epoch.Foreach(func(sys byte, msm *gnss.RtcmMsm) {
					if rval == nil {
						rval = gnss.NewRinexHeader()
					}
					rval.CheckAddMsmHeader(msm.Header)
				})
				j++
				return true
			})
			if j >= 60 { // 60个历元
				return false // 不需要再循环
			}
		}
		return true
	})

	return rval
}

func (this *GnssRinexObs3xWriterAfter) InputRtcmMSM(obs *gnss.RtcmMsm) bool {
	if !this.Filter_Msm_EndT.IsZero() { // 截止时间直接退出
		if obs.MsmTime.After(this.Filter_Msm_EndT) {
			return false
		}
	}
	if !this.Filter_Msm_StartT.IsZero() {
		if obs.MsmTime.Before(this.Filter_Msm_StartT) {
			return true
		}
	}
	if this.msmInput == nil {
		this.msmInput = gnss.NewRtcmMsmEpochInput()

		this.msmInput.OnRtcmMsmEpochEnd = func(epoch *gnss.RtcmMsmEpoch) { // 收到一个历元
			this.checkInputEpoch(epoch)
		}
	}
	//调试
	//if !this.Filter_Msm_StartT.IsZero(){
	//	if obs.MsmTime.After(this.Filter_Msm_StartT) {
	//		if obs.Sys == gnss.SYS_GLO{
	//			obs.Sys = gnss.SYS_GLO
	//		}
	//	}
	//}

	if this.firstRecvRtcmT.IsZero() {
		this.firstRecvRtcmT = time.Now()
	}

	this.msmInput.InputMsm(obs, func() bool {
		return this.checkFilterNaviSys(obs) && this.checkFilterMsm(obs)
	})

	return true

}
